package com.lagou.client;

import com.alibaba.fastjson.JSONObject;
import com.lagou.common.JSONSerializer;
import com.lagou.common.RpcEncoder;
import com.lagou.pojo.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.I0Itec.zkclient.ZkClient;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RpcConsumer {

    //创建线程池对象
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static List<Integer> ports = new ArrayList<Integer>();

    private static List<String> hosts = new ArrayList<String>();

    private static List<Channel> channelList = new ArrayList<>();

    ZkClient zkClient = new ZkClient("127.0.0.1:2181", 5000);

    public static String result = "";

    //1.创建一个代理对象 providerName：UserService#sayHello are you ok?
    public Object createProxy(final Class<?> serviceClass,final RpcRequest rpcRequest){
        //借助JDK动态代理生成代理对象
        return  Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

                if(channelList.size() == 0){
                    initClient();
                }

                Channel channel = getChannel(channelList);
                channel.writeAndFlush(rpcRequest);

                return result;
            }
        });


    }



    //2.初始化netty客户端
    public static  void initClient() throws InterruptedException {
        //获取连接信息
        getConnectInfo();

        channelList = new ArrayList<>();

        for (int i = 0; i < hosts.size(); i++) {

            String host = hosts.get(i);
            int port = ports.get(i);

            UserClientHandler userClientHandler = new UserClientHandler();

            EventLoopGroup group = new NioEventLoopGroup();

            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY,true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new UserClientHandler());

                        }
                    });


            Channel channel = bootstrap.connect(host, port).channel();
            channelList.add(channel);

        }

    }

    /**
     * 从zk中获取连接信息
     */
    public static void getConnectInfo(){

        ZkClient zkClient = new ZkClient("127.0.0.1:2181", 5000);
        List<String> list = zkClient.getChildren("/server");

        ports = new ArrayList<Integer>();
        hosts = new ArrayList<String>();

        for (String str : list) {
            String host = str.split(":")[0];
            String port = str.split(":")[1];
            hosts.add(host);
            ports.add(Integer.valueOf(port));
        }

    }

    /**
     * 负载均衡 获取最后一次响应时间短的
     * @param channelList
     * @return
     */
    public Channel getChannel(List<Channel> channelList){

        ZkClient zkClient = new ZkClient("127.0.0.1:2181", 15000);

        Map<String,Integer> serverTimeMap = new HashMap<String,Integer>();
        Map<String,Channel> channelMap = new HashMap<String,Channel>();

        List<Integer> runTimeList = new ArrayList<Integer>();
        List<String> serverNameList = new ArrayList<String>();

        for (Channel channel : channelList) {

            InetSocketAddress inetSocketAddress= (InetSocketAddress)channel.remoteAddress();
            String serverName = "/server/"+inetSocketAddress.getAddress().getHostAddress()+":"+inetSocketAddress.getPort();

            channelMap.put(serverName,channel);

            Object o = zkClient.readData(serverName);

            //如果没有时间新启动的默认为0
            if(o != null){

                String jsonStr = o.toString();
                JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                Integer runTime = Integer.valueOf(jsonObject.get("runTime").toString());
                serverTimeMap.put(serverName,runTime);
                runTimeList.add(runTime);
            }else{
                serverTimeMap.put(serverName,0);
                runTimeList.add(0);
            }
        }

        if(serverTimeMap.size() != 0 ){
            
            Integer min = Collections.min(runTimeList);

            serverTimeMap.forEach((key, value) -> {
                if(value.equals(min)){
                    serverNameList.add(key);
                }
            });
            //如果时间有一致的随机
            Random random = new Random();

            int n = random.nextInt(serverNameList.size());
            String key = serverNameList.get(n);
            Channel channel = channelMap.get(key);

            return channel;
        }

        return null;
    }

}
